package concurrencyPractice.chapter07;

import java.math.BigInteger;
import java.util.concurrent.*;

/**
 * @author xinya
 * @version 1.0
 * @project javacore
 * @description 7-3 不可靠的取消操作将把生产者置于阻塞的操作中
 * @date 2024/1/4 16:45:47
 */
class BrokenPrimerProducer extends Thread {
    private final BlockingQueue<BigInteger> queue;
    private volatile boolean canceled = false;

    BrokenPrimerProducer(BlockingQueue<BigInteger> queue) {
        this.queue = queue;
    }

    public void run() {
        try {
            BigInteger p = BigInteger.ONE;
            while (!canceled) {
                queue.put(p = p.nextProbablePrime());
            }
        } catch (InterruptedException consumed) {

        }
    }

    public void cancel() {
        canceled = true;
    }

    void consumePrimes() throws InterruptedException {
        BlockingQueue<BigInteger> primes = new LinkedBlockingQueue<>();
        BrokenPrimerProducer producer = new BrokenPrimerProducer(primes);
        producer.start();
        try {
            while (needMorePrimes()) {
                consume(primes.take());
            }
        } finally {
            producer.cancel();
        }
    }

    boolean needMorePrimes() {
        return false;
    }

    void consume(BigInteger p) {

    }
}

/**
 * 7-5 通过中断取消
 */
public class PrimeProducer extends Thread {

    private final BlockingDeque<BigInteger> queue;
    private volatile boolean canceled = false;

    PrimeProducer(BlockingDeque<BigInteger> queue) {
        this.queue = queue;
    }

    public void run() {
        try {
            BigInteger p = BigInteger.ONE;
            while (!Thread.currentThread().isInterrupted()) {
                queue.put(p = p.nextProbablePrime());
            }
        } catch (InterruptedException consumed) {

        }
    }

    public void cancel() {
        interrupt();
    }

    /**
     * 7-7 不可取消的任务在退出前恢复中断
     *
     * @param queue
     * @return
     */
    public Task getNextTask(BlockingQueue<Task> queue) {
        boolean interrupted = false;
        try {
            while (true) {
                try {
                    return queue.take();
                } catch (InterruptedException e) {
                    interrupted = true;
                    // 重新尝试
                }
            }
        } finally {
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 7-8 在外部线程中安排中断（NOT）
     */
    private static final ScheduledExecutorService cancelExec = new ScheduledThreadPoolExecutor(10);
    public static void timedRun(Runnable r, long timeout, TimeUnit unit) {
        final Thread taskThread = Thread.currentThread();
        cancelExec.schedule(new Runnable() {
            @Override
            public void run() {
                taskThread.interrupt();
            }
        }, timeout, unit);
        r.run();
    }

    /** 7-9 在专门的线程中中断任务
     *
     * @param r
     * @param timeout
     * @param unit
     * @throws InterruptedException
     */
    public static void timedRun2(final Runnable r, long timeout, TimeUnit unit) throws InterruptedException {
        class RethrowableTask implements Runnable {
            private volatile Throwable t;

            @Override
            public void run() {
                try {
                    r.run();
                } catch (Throwable t) {
                    this.t = t;
                }
            }
            void rethrow () {
                if(t!= null) {
//                    throw launderThrowable(r);
                }
            }
        }

        RethrowableTask task = new RethrowableTask();
        final Thread taskThread = new Thread();
        taskThread.start();
        cancelExec.schedule(new Runnable() {
            @Override
            public void run() {
                taskThread.interrupt();
            }
        }, timeout, unit);
        taskThread.join(unit.toMillis(timeout));
        task.rethrow(); // 检查再次抛出
    }

    private static final ExecutorService taskExec = Executors.newFixedThreadPool(10);
    public static void timedRun3(final Runnable r, long timeout, TimeUnit unit) throws InterruptedException {
        Future<?> task = taskExec.submit(r);
        try {
            task.get(timeout, unit);
        } catch (TimeoutException e) {
            // 任务将被取消
        }catch (ExecutionException e) {
            // 如果任务抛出了异常那么冲洗抛出
//            throw launderThrowable(e.getCause());
        } finally {
            // 如果任务结束，执行取消不会有任何影响
            // 如果任务正在运行，那么将被中断
            task.cancel(true);
        }
    }

}

class Task {

}


